package com.dahuan.stream;

import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;

public class LogProcessor implements Processor<byte[],byte[]> {

    private ProcessorContext context;

    @Override
    public void init(ProcessorContext processorContext) {

        context = processorContext;
    }

    @Override
    public void process(byte[] key, byte[] value) {

        //TODO 获取一行数据
        String line = new String( value );

        //TODO 如果包含“>>>”则只保留该标记后面的内容
        if (line.contains( ">>>" )){
            line = line.split( ">>>" )[1].trim();
            //TODO 输出到下一个topic
            context.forward( "LogProcessor".getBytes(),line.getBytes() );
        }else {
            context.forward( "LogProcessor".getBytes(),line.getBytes() );
        }

        
      /*  //TODO 去除 “ >>> ”

        line.replaceAll(",","");

        value = line.getBytes();

        context.forward( key,value );*/

    }

    @Override
    public void punctuate(long timestamp) {

    }

    @Override
    public void close() {

    }
}
